
#include <apr-1/apr_strings.h>
#include "../crypt/CryptEChain.h"
#include "../crypt/RSAKeyPair.h"
#include "../utils/Hash.h"
#include "../utils/StringUtils.h"
#include "../utils/SerialBlockSource.h"
#include "../net/Broker4.h"
#include "../xml/DOMDocument.h"
#include "../xml/XMLDescriptor.h"
#include "Broker4.h"

std::vector<char> *Broker4::SimpleCryptDataChunk(const char *cname, const char *sname, 
   KeyPair *rsa, KeyPair *signing, const char *datachunk, int length, int dodecrypt, int docompress, apr_pool_t *mp, int *trusted)
{
   APRBucketBrigade *decryptbb = Broker4::CryptDataChunkToBucket(cname, sname, rsa, signing, datachunk, length, dodecrypt, docompress, mp, trusted);
   if (!decryptbb)
   {
      errprint("Could not decrypt data chunk to bucket");
      return NULL;
   }
   std::vector<char> *largefile = new std::vector<char>();
   apr_size_t reallen = decryptbb->realLength();
   largefile->reserve(reallen);
   if (reallen > 0)
   {
      char *alength = (char *)malloc(sizeof(char)*(reallen+1));
      apr_size_t gotreal = decryptbb->popFromStream(alength, reallen);
      for (int ii = 0; ii < (int)gotreal; ii++)
         largefile->push_back(alength[ii]);
      if (gotreal != reallen)
      {
         errprint("Could not get all of the bucket brigade");
         free(alength);
         delete largefile;
         return NULL;
      }
      free(alength);
   }
   delete decryptbb;
   return largefile;
}

APRBucketBrigade *Broker4::CryptDataChunkToBucket(const char *cname, const char *sname,
   KeyPair *rsa, KeyPair *signing, const char *datachunk, int length, int dodecrypt, int docompress, apr_pool_t *mp, int *trusted)
{
   CryptEChain *crypty = new CryptEChain();
   crypty->buildBasicChain2(cname, sname, rsa, signing, dodecrypt, docompress, length);
   APRBucketBrigade *bucketbrgde = new APRBucketBrigade(mp);
   int smallblock = (ENV_READ_BUFFER > (length/2)) ? (length/2) : ENV_READ_BUFFER;
   for (int k = 0; k < length; k += smallblock)
   {
      int actualamt = ((length - k) < smallblock) ? length - k : smallblock;
      int bytes = bucketbrgde->writeToStream(&(datachunk[k]), actualamt);
      if (bytes == 0)
      {
         errprint("push to bucket brigade failed - incomplete data will be sent");
         break;
      }
      k -= (smallblock - bytes);
      // Check bytes, seriously!
   }
   bucketbrgde->writeEOS();
   APRBucketBrigade *outputbrgde = new APRBucketBrigade(mp);
   int prociter = 0;
   while (prociter == 0)
      prociter = crypty->processIteration(bucketbrgde, outputbrgde);
   delete bucketbrgde;
   if (trusted) *trusted = crypty->isTrusted();
   delete crypty;
   if (prociter == -1)
   {
      delete outputbrgde;
      return NULL;
   }
   return outputbrgde;
}

std::vector<char> *Broker4::ReadIntoBuffer(APRBucketBrigade *aprbb, int len, APRMutex *waiting_block)
{
   int actualreq;
   std::vector<char> *returned = new std::vector<char>();
   char *buf = (char *)malloc(sizeof(char)*MAX_BLOCK_SIZE);
   while (1)
   {
      actualreq = (MAX_BLOCK_SIZE < (len - returned->size())) ? MAX_BLOCK_SIZE : (len - returned->size());
      if (len == -1) actualreq = MAX_BLOCK_SIZE;
      apr_size_t len2 = aprbb->popFromStream(buf, actualreq);
      if (len2 == 0) if (waiting_block) waiting_block->wait();
      for (int i = 0; i < (int)len2; i++)
         returned->push_back(buf[i]);
      if (aprbb->isClosed()) break;
      if (returned->size() == len) break;
   }
   free(buf);
   return returned;
}

apr_size_t Broker4::GetContentLength(const apr_table_t *htable)
{
   const char *clength_str = apr_table_get(htable, "Content-Length");
   if (clength_str)
   {
      apr_size_t clength_true = 0;
      sscanf(clength_str, "%" APR_SIZE_T_FMT, &clength_true);
      return clength_true;
   }
   return (apr_size_t)-1;
}

apr_table_t *Broker4::GetTable(char *buffer, apr_size_t clength, const char *separator, const char *eol, apr_pool_t *mp)
{
   apr_table_t *htable = apr_table_make(mp, MAX_TABLE_ENTS);
   char *remainder = NULL;
   char *token = apr_strtok(buffer, separator, &remainder);
   while (token != NULL)
   {
      char *tkey = token;
      char *tvalue = apr_strtok(NULL, eol, &remainder);
      if (tvalue != NULL)
      {
         apr_table_set(htable, tkey, tvalue);
      } else break;
      token = apr_strtok(NULL, separator, &remainder);
   }
   return htable;
}

apr_table_t *Broker4::GetTable(APRBucketBrigade *aprbb, apr_size_t clength, const char *separator, const char *eol, apr_pool_t *mp, APRMutex *waiting_block)
{
   int isfirst = 1;
   apr_table_t *htable = apr_table_make(mp, MAX_TABLE_ENTS);
   apr_size_t consumedbytes = 0;
   apr_size_t length_last_time = 0;
   char *linebuffer = (char *)malloc(sizeof(char)*MAX_BLOCK_SIZE);
   linebuffer[0] = '\0';
   while (!(aprbb->isClosed()))
   {
      apr_size_t length_this_time = aprbb->realLength();
      if (!waiting_block) length_this_time = length_last_time + 1;
      if ((length_last_time != length_this_time) ||
          ((length_this_time + consumedbytes) >= clength))
      {
         int realbytes = aprbb->readFromStream(linebuffer, MAX_BLOCK_SIZE - 1);
         linebuffer[realbytes] = '\0';
         char *k = strstr(linebuffer, eol);
         if ((!k) && ((length_this_time + consumedbytes) >= clength))
            k = linebuffer + strlen(linebuffer);
         if (k)
         {
            if (k == linebuffer)
            {
               aprbb->removeFromStream(strlen(eol));
               free(linebuffer);
               return htable;
            } else
            {
               int vallen = (k - linebuffer) + strlen(eol);
               char *newheader = (char *)apr_palloc(mp, sizeof(char)*(vallen+1));
               memcpy(newheader, linebuffer, k - linebuffer);
               newheader[k - linebuffer] = '\0';
               char *tremainder = NULL;
               char *bits = apr_strtok(newheader, separator, &tremainder);
               // isfirst is a fudge to remove extra characters in later headers and
               // make the first header line in keeping (somewhat) with the rest
               if (tremainder[0] == ' ') tremainder++;
               apr_table_set(htable, bits, tremainder);
               aprbb->removeFromStream(vallen);
               consumedbytes += vallen;
               isfirst = 0;
            }
         } else
         {
            if (length_this_time > (MAX_BLOCK_SIZE - 1))
            {
               aprbb->writeEOS();
               errprint("data too big");
               free(linebuffer);
               return NULL;
            }
         }
         length_last_time = length_this_time;
      } else if (waiting_block) waiting_block->wait();
   }
   free(linebuffer);
   return htable;
}

int Broker4::WriteEncryptedMessageToBucket(APRBucketBrigade *aprbb, EncMessage *msg, KeyDB *kdb)
{
   
}

int Broker4::WriteEncMessageToBB(APRBucketBrigade *aprbb, EncMessage *msg, KeyDB *kdb)
{
   if (!msg)
   {
      errprint("Did not pass a suitable EncMessage class");
      return 1;
   }
   // Get keys
   KeyPair *fromkey = kdb->GetPrivateKeyByName(msg->GetSender());
   if (fromkey == NULL)
      errprint("Don't have the keys to send a message from '%s'", msg->GetSender());
   if ((fromkey) && (!(msg->ToBeSigned())))
   {
      delete fromkey;
      fromkey = NULL;
   }
   KeyPair *tokey = kdb->GetPublicKeyByName(msg->GetRecipient());
   if (tokey == NULL)
   {
      errprint("Don't have the keys to send a message to '%s'", msg->GetRecipient());
//      errprint("headers from offending message: %s", msg->headers);
      delete fromkey;
      return 1;
   }
   // Build encrypted headers
   std::vector<char *> *encrypted_headers = new std::vector<char *>();
   const char *stream = msg->GetHeaderStream();
   if (stream == NULL)
   {
      errprint("No headers specified in this outgoing message?");
      delete fromkey;
      delete tokey;
      delete encrypted_headers;
      return 1;
   }
   char *stream_dup = (char *)malloc(sizeof(char)*(strlen(stream)+1));
   strcpy(stream_dup, stream);
   encrypted_headers->push_back(stream_dup);
   char *payloads = StringUtils::varprintf("Payload-Count: %d" CRLF_STR, msg->GetHowManyPayloads());
   encrypted_headers->push_back(payloads);
   int sum_of_lengths = 0;
   for (int i = 0; i < msg->GetHowManyPayloads(); i++)
   {
      int plength = msg->GetPayloadLength(i);
      if (plength < 0)
         errprint("Payload %d of invalid length %d", i, plength);
      char *uencc = StringUtils::varprintf("UContent-Length-%d: %d" CRLF_STR, i, plength);
      encrypted_headers->push_back(uencc);
      sum_of_lengths += plength;
   }
   // Build content
   char *bigbuffer = (char *)malloc(sizeof(char)*sum_of_lengths);
   int reset_sum = 0;
   for (int i = 0; i < msg->GetHowManyPayloads(); i++)
   {
      int plength = msg->GetPayloadLength(i);
      memcpy(&(bigbuffer[reset_sum]), msg->GetPayload(i), sizeof(char)*plength);
      reset_sum += plength;
   }
   // Encrypt content
   std::vector<char> *encryptedcontent = NULL;
   if (sum_of_lengths > 0)
   {
      encryptedcontent = Broker4::SimpleCryptDataChunk(msg->GetRecipient(), msg->GetSender(), tokey, fromkey, bigbuffer, sum_of_lengths, 0, 1, msg->GetEncMessageAssociatedMemoryPool(), NULL);
      if ((encryptedcontent) && (encryptedcontent->size() == 0))
      {
         errprint("content encryption failed: no encrypted output");
         delete fromkey;
         delete tokey;
         StringUtils::destroyStringList(encrypted_headers);
         delete encryptedcontent;
         free(bigbuffer);
         return 1;
      }
   }
   free(bigbuffer);
   // Find real content length - after encryption - and put this in the header
   if (sum_of_lengths > 0)
   {
      char *rsze = StringUtils::varprintf("Content-Length: %d" CRLF_STR, encryptedcontent->size());
      encrypted_headers->push_back(rsze);
   }
   char *oops = StringUtils::varprintf(CRLF_STR);
   encrypted_headers->push_back(oops);
   // Header complete - flatten it
   char *fullencheader = StringUtils::flattenStringList(encrypted_headers);
   StringUtils::destroyStringList(encrypted_headers);
   std::vector<char> *encryptedheader =
      Broker4::SimpleCryptDataChunk(msg->GetRecipient(), msg->GetSender(), tokey, fromkey, fullencheader, strlen(fullencheader), 0, 0, msg->GetEncMessageAssociatedMemoryPool(), NULL);
   if ((encryptedheader) && (encryptedheader->size() == 0))
   {
      errprint("header encryption failed: no encrypted output");
      delete fromkey;
      delete tokey;
      if (encryptedcontent) delete encryptedcontent;
      delete encryptedheader;
      free(fullencheader);
      return 1;
   }
   free(fullencheader);
   delete fromkey;
   delete tokey;
   // Complete now for the plain text header
   char *ptheader = StringUtils::varprintf(
      "FROM %s" CRLF_STR
      "TO %s" CRLF_STR
      "Content-Length: %d" CRLF_STR CRLF_STR,
      msg->GetSender(), msg->GetRecipient(), encryptedheader->size());
   apr_size_t sentlen = 0;
   sentlen = aprbb->writeToStream(ptheader, strlen(ptheader));
   if (sentlen != strlen(ptheader))
   {
      errprint("Could not send plain headers (%d out of %d bytes sent)", sentlen, strlen(ptheader));
      free(ptheader);
      if (encryptedcontent) delete encryptedcontent;
      delete encryptedheader;
      return 1;
   }
   free(ptheader);
   sentlen = aprbb->writeToStream(&(encryptedheader->at(0)), encryptedheader->size());
   if (sentlen != encryptedheader->size())
   {
      errprint("Could not send encrypted headers (%d out of %d bytes sent)", sentlen, encryptedheader->size());
      if (encryptedcontent) delete encryptedcontent;
      delete encryptedheader;
      return 1;
   }
   delete encryptedheader;
   if (encryptedcontent)
   {
      sentlen = aprbb->writeToStream(&(encryptedcontent->at(0)), encryptedcontent->size());
      if (sentlen != encryptedcontent->size())
      {
         errprint("Could not send encrypted content (%d out of %d bytes sent)", sentlen, encryptedcontent->size());
         delete encryptedcontent;
         return 1;
      }
      delete encryptedcontent;
   }
   return 0;
}

int Broker4::ReadEncMessageFromBB(APRBucketBrigade *aprbb, EncMessage *msg, KeyDB *kdb, APRMutex *waiting_block)
{
   if (!msg)
   {
      errprint("Did not pass a suitable EncMessage class");
      return 1;
   }
   apr_table_t *clear_headers_table = Broker4::GetTable(aprbb, (apr_size_t)-1, ": ", CRLF_STR, msg->GetEncMessageAssociatedMemoryPool(), waiting_block);
   if (clear_headers_table == NULL)
   {
      errprint("Headers too big or received malformed message");
      return 1;
   }
   const char *from_static;
   const char *to_static;
   if ((from_static = apr_table_get(clear_headers_table, "FROM")) == NULL)
   {
      errprint("Could not find the FROM header");
      return 1;
   }
   if ((to_static = apr_table_get(clear_headers_table, "TO")) == NULL)
   {
      errprint("Could not find the TO header");
      return 1;
   }
   msg->SetRecipient(to_static);
   msg->SetSender(from_static);
   // Get keys
   KeyPair *fromkey = kdb->GetPublicKeyByName(msg->GetSender());
   // Username beginning with a dot is an untrusted temporary key that is not used for authentication
   if ((fromkey == NULL) && (!((msg->GetSender())[0] == '.')))
      errprint("Don't have the keys to authenticate a message from '%s'", msg->GetSender());
   KeyPair *tokey = kdb->GetPrivateKeyByName(msg->GetRecipient());
   if (tokey == NULL)
   {
      errprint("Don't have the keys to receive a message addressed to '%s'", msg->GetRecipient());
      msg->Reset();
      return 1;
   }
   apr_size_t first_block_size = Broker4::GetContentLength(clear_headers_table);
   std::vector<char> *encrypted_headers_flat = Broker4::ReadIntoBuffer(aprbb, first_block_size, waiting_block);
   if (first_block_size != (encrypted_headers_flat->size()))
   {
      errprint("Expected encrypted headers size %d bytes, obtained size %d bytes", first_block_size, encrypted_headers_flat->size());
      msg->Reset();
      delete encrypted_headers_flat;
      if (fromkey) delete fromkey;
      if (tokey) delete tokey;
      return 1;
   }
   int trusted_a = 0;
   std::vector<char> *decrypted_headers =
      Broker4::SimpleCryptDataChunk(msg->GetRecipient(), msg->GetSender(), tokey, fromkey, &(encrypted_headers_flat->at(0)), encrypted_headers_flat->size(), 1, 0, msg->GetEncMessageAssociatedMemoryPool(), &(trusted_a));
   delete encrypted_headers_flat;
   if ((!decrypted_headers) || (decrypted_headers->size() <= 0))
   {
      errprint("Failed to decrypt headers of message from '%s' to '%s'", from_static, to_static);
      msg->Reset();
      if (decrypted_headers) delete decrypted_headers;
      if (fromkey) delete fromkey;
      if (tokey) delete tokey;
      return 1;
   }
   decrypted_headers->push_back('\0');
   msg->ParseHeaderStream(&(decrypted_headers->at(0)));
   const apr_table_t *decrypted_headers_table = msg->GetHeaderTable();
   delete decrypted_headers;
   const char *nopayloads_string;
   if ((nopayloads_string = apr_table_get(decrypted_headers_table, "Payload-Count")) == NULL)
   {
      errprint("No Payload-Count header in decrypted headers");
      errprint("%s", msg->GetHeaderStream());
      msg->Reset();
      if (fromkey) delete fromkey;
      if (tokey) delete tokey;
      return 1;
   }
   int nopayloads = atoi(nopayloads_string);
   if (nopayloads < 0)
   {
      errprint("Invalid number of payloads");
      msg->Reset();
      if (fromkey) delete fromkey;
      if (tokey) delete tokey;
      return 1;
   } else if (nopayloads == 0)
   {
      if (trusted_a) msg->SetWasSignedFlag();
      // It's okay to not have a payload!
      if (fromkey) delete fromkey;
      if (tokey) delete tokey;
      return 0;
   }
   apr_size_t second_block_size = Broker4::GetContentLength(decrypted_headers_table);
   std::vector<char> *encrypted_content_flat = Broker4::ReadIntoBuffer(aprbb, second_block_size, waiting_block);
   if (second_block_size != (encrypted_content_flat->size()))
   {
      errprint("Expected encrypted content size %d bytes, obtained size %d bytes", second_block_size, encrypted_content_flat->size());
      msg->Reset();
      delete encrypted_content_flat;
      if (fromkey) delete fromkey;
      if (tokey) delete tokey;
      return 1;
   }
   int trusted_b = 0;
   std::vector<char> *decrypted_contents = Broker4::SimpleCryptDataChunk(msg->GetRecipient(), msg->GetSender(), tokey, fromkey, &(encrypted_content_flat->at(0)), encrypted_content_flat->size(), 1, 1, msg->GetEncMessageAssociatedMemoryPool(), &(trusted_b));
   if (fromkey) delete fromkey;
   if (tokey) delete tokey;
   delete encrypted_content_flat;
   if ((!decrypted_contents) || (decrypted_contents->size() == 0))
   {
      errprint("Failed to decrypt headers of message from '%s' to '%s'", msg->GetSender(), msg->GetRecipient());
      msg->Reset();
      if (decrypted_contents) delete decrypted_contents;
      return 1;
   }
   // Whether both the head and the body are trusted!
   if ((trusted_a) && (trusted_b)) msg->SetWasSignedFlag();
   int alllength = 0;
   for (int i = 0; i < nopayloads; i++)
   {
      char *theader = StringUtils::varprintf("UContent-Length-%d", i);
      const char *thispayload_length_string;
      if ((thispayload_length_string = apr_table_get(decrypted_headers_table, theader)) == NULL)
      {
         errprint("Header '%s' for the length of payload %d does not exist", theader, i);
         msg->Reset();
         if (theader) free(theader);
         delete decrypted_contents;
         return 1;
      }
      free(theader);
      int thispayload_length = atoi(thispayload_length_string);
      if ((thispayload_length + alllength) > decrypted_contents->size())
      {
         errprint("Content length for payload %d does not make sense (%d + %d > %d)", thispayload_length, alllength, decrypted_contents->size());
         msg->Reset();
         delete decrypted_contents;
         return 1;
      }
      msg->AddPayloadBuffer(&(decrypted_contents->at(alllength)), thispayload_length);
      alllength += thispayload_length;
   }
   delete decrypted_contents;
   return 0;
}

SSHChain *Broker4::GetClusterConnection(AllDBs *dbs, KeyDB *sshkeydb, const char *final_user, const char *sshkeychainpass, const char *clustertoken, apr_pool_t *mp)
{
   // pick the first one and chain to it...
   int fingerprintfailed = 0;
   char *remainder = NULL;
   SSHChain *tchain = NULL;
   std::vector<char *> *blacklist = new std::vector<char *>();
   while ((clustertoken != NULL) && (tchain == NULL))
   {
      std::vector<char *> *hostpath = dbs->clustersdb->GetPathFromSourceToCluster(final_user, clustertoken, blacklist);
      while ((hostpath != NULL) && (tchain == NULL))
      {
         tchain = new SSHChain(dbs->clustersdb, mp);
         for (int i = 0; i < hostpath->size(); i++)
            tchain->AddToSSHChain(final_user, hostpath->at(i));
         tchain->SetAuthenticatorKeys(sshkeydb, final_user, sshkeychainpass);
         int didfail;
         SSHClient *scli = tchain->ActivateSSHChain(&didfail);
         if (didfail)
         {
            char *failedhost = tchain->GetFailedLinkHostName();
            errprint("failed to access '%s', blacklisting...", failedhost);
            blacklist->push_back(failedhost);
            tchain->DustForFingerprints(dbs->pendingdb);
            fingerprintfailed = tchain->WasFingerprintProblem();
            delete tchain;
            tchain = NULL;
            StringUtils::destroyStringList(hostpath);
            hostpath = dbs->clustersdb->GetPathFromSourceToCluster(final_user, clustertoken, blacklist);
         }
      }
      if (hostpath) StringUtils::destroyStringList(hostpath);
   }
   if (blacklist) StringUtils::destroyStringList(blacklist);
   return tchain;
}

char *Broker4::GetClusterListForWorkStore(AllDBs *dbs, KeyDB *sshkeydb, const char *store_id, const char *clusterlist, const char *sshkeychainpass, int *success, apr_pool_t *mp)
{
   //  0 - no error
   // -1 - general error
   // -2 - user has no clusters
   // -3 - user cannot access specified clusters
   // -4 - cluster capabilities do not match the capabilities required by the job
   // -5 - cannot contact any appropriate clusters
   if ((store_id == NULL) || (strlen(store_id) == 0))
   {
      errprint("cannot start a work store with %s name", (store_id) ? "an empty" : "a NULL");
      *success = -1;
      return NULL;
   }
   char *final_user = dbs->actorsdb->LookupUltimateUser(store_id);
   if ((final_user == NULL) || (strlen(final_user) == 0))
   {
      errprint("could not find a user for the work store");
      if (final_user) free(final_user);
      *success = -1;
      return NULL;
   }
   char *access_clusters = dbs->clustersdb->GetListOfClustersForUser(final_user);
   if ((access_clusters == NULL) || (strlen(access_clusters) == 0))
   {
      errprint("can't start job '%s' with no cluster access!", store_id);
      free(final_user);
      if (access_clusters) free(access_clusters);
      *success = -2;
      return NULL;
   }
   char *str_clusterids = NULL;
   if ((clusterlist == NULL) || (strlen(clusterlist) == 0))
   {
      str_clusterids = (char *)malloc(sizeof(char)*(strlen(access_clusters)+1));
      strcpy(str_clusterids, access_clusters);
   } else
   {
      str_clusterids = StringUtils::intersectCSL(clusterlist, access_clusters);
   }
   free(access_clusters);
   if ((str_clusterids == NULL) || (strlen(str_clusterids) == 0))
   {
      errprint("can't start job '%s' with none of the provided clusters accessible!", store_id);
      if (str_clusterids) free(str_clusterids);
      free(final_user);
      *success = -3;
      return NULL;
   }
//   XMLDescriptor *xmdescr = new XMLDescriptor(dbs->keydb, 
   // load up XML descriptor and check capabilities - skip this for now
   // Should compare cluster caps here.
   char *remainder = NULL;
   char *clustertoken = apr_strtok(str_clusterids, ",", &remainder);
   char *varpnt = StringUtils::varprintf("");
   while (clustertoken != NULL)
   {
      SSHChain *workingchain = Broker4::GetClusterConnection(dbs, sshkeydb, final_user, sshkeychainpass, clustertoken, mp);
      if (workingchain)
      {
         char *oldvarpnt = varpnt;
         varpnt = StringUtils::varprintf((strlen(varpnt) == 0) ? "%s%s" : "%s,%s", oldvarpnt, clustertoken);
         free(oldvarpnt);
         delete workingchain;
      }
      clustertoken = apr_strtok(NULL, ",", &remainder);
   }
   free(str_clusterids);
   str_clusterids = varpnt; 
   if ((str_clusterids == NULL) || (strlen(str_clusterids) == 0))
   {
      errprint("can't start job '%s' with none of the provided clusters contactable!", store_id);
      if (str_clusterids) free(str_clusterids);
      free(final_user);
      *success = -5;
      return NULL;
   }
   free(final_user);
   *success = 0;
   return str_clusterids;
}

int Broker4::TryToStartWorkStore(AllDBs *dbs, const char *store_id, int instances, int maxtimesecs, const char *sshkeychainpass, const char *clusterlist, apr_pool_t *mp)
{
   // -2 - general error
   // -1 - no appropriate cluster
   int success;
   char *realclusterchain = GetClusterListForWorkStore(dbs, dbs->sshkeydb, store_id, clusterlist, sshkeychainpass, &success, mp);
   if ((realclusterchain == NULL) || (strlen(realclusterchain) == 0))
   {
      // possible more details as to why there are none will be in the success variable
      if (realclusterchain) free(realclusterchain);
      return 0;
   }
   char *sso_userid = dbs->actorsdb->LookupUltimateUser(store_id);
   if ((sso_userid == NULL) || (strlen(sso_userid) == 0))
   {
      errprint("user id is %s", (sso_userid) ? "NULL" : "empty");
      if (realclusterchain) free(realclusterchain);
      if (sso_userid) free(sso_userid);
      return 0;
   }
   std::vector<char *> *usersshkeys = dbs->clustersdb->getSetOfActiveKeysForUser(sso_userid);
   if ((usersshkeys == NULL) || (usersshkeys->size() == 0))
   {
      errprint("no user SSH keys found");
      if (realclusterchain) free(realclusterchain);
      if (sso_userid) free(sso_userid);
      if (usersshkeys) delete usersshkeys;
      return 0;
   }
   for (int i = 0; i < usersshkeys->size(); i++)
   {
      // Transplant SSH keys
      KeyPair *pubkey = dbs->sshkeydb->GetPublicKeyByName(usersshkeys->at(i));
      KeyPair *prvkey = dbs->sshkeydb->GetPrivateKeyByName(usersshkeys->at(i), sshkeychainpass);
      char *sshkeyowner = StringUtils::varprintf(".ssh_%s", sso_userid);
      if (!(dbs->keydb->CheckForPublicKeyByName(usersshkeys->at(i))))
      {
         dbs->keydb->AddPublicKey(usersshkeys->at(i), pubkey, sshkeyowner);
         dbs->keydb->AddPrivateKeyWithPassword(usersshkeys->at(i), dbs->keydb->WIPpassword, prvkey, sshkeyowner);
      }
      free(sshkeyowner);
      dbs->keydb->KeyPairNowUsedByActor(usersshkeys->at(i));
      dbs->keydb->MarkKeyPairAsLatestButExpendableWhenZero(usersshkeys->at(i));
      dbs->actorsdb->AddWorkStoreKeyDependency(store_id, usersshkeys->at(i));
      delete pubkey;
      delete prvkey;
   }
   // Have verified that there are places for this to run... start it.
   dbs->actorsdb->CreateWorkFromFileStoreID(store_id, maxtimesecs, instances);
   dbs->actorsdb->WorkStoreSetStarted(store_id);
   free(realclusterchain);
   if (sso_userid) free(sso_userid);
   if (usersshkeys) StringUtils::destroyStringList(usersshkeys);
   return 1;
}

void Broker4::TryToStartUnresourcedWork(AllDBs *dbs, const char *exthost, int extport, apr_pool_t *mp)
{
   std::vector<char *> *unresourced_work = dbs->actorsdb->sqlScanAllTextf("select name,file_store,tmax from 'workdata' where computename='';");
   if ((unresourced_work == NULL) || (unresourced_work->size() == 0))
   {
      if (unresourced_work) delete unresourced_work;
      return;
   } else
   {
      errprint("found %d piece(s) of unresourced work!", unresourced_work->size() / 3);
      for (int i = 0; i < (unresourced_work->size() / 3); i++)
      {
         char *woouser = StringUtils::mpprintf(mp, ".compute_%s", unresourced_work->at(i*3+1));
         char *final_user = dbs->actorsdb->LookupUltimateUser(unresourced_work->at(i*3+1));
         RSAKeyPair *rsakp = new RSAKeyPair();
         rsakp->GenerateKeyPair(KEY_BITS);
         // Give it a compute key name
         char *newuuid = Hash::makeHexUUID_PlusUniq2(24, dbs->keydb, dbs->sshkeydb);
         dbs->keydb->AddPublicKey(newuuid, (KeyPair *)rsakp, woouser);
         dbs->keydb->AddPrivateKeyWithPassword(newuuid, dbs->keydb->WIPpassword, (KeyPair *)rsakp, woouser);
         int success = 0;
         char *clusterlist = Broker4::GetClusterListForWorkStore(dbs, dbs->keydb, unresourced_work->at(i*3+1), NULL, dbs->keydb->WIPpassword, &success, mp);
         int clustercount = 0;
         char *remainder = NULL;
         char *clustertoken = apr_strtok(clusterlist, ",", &remainder);
         std::vector<char *> *sclusterlist = new std::vector<char *>();
         while (clustertoken != NULL)
         {
            sclusterlist->push_back(clustertoken);
            clustercount++;
            clustertoken = apr_strtok(NULL, ",", &remainder);
         }
         int nodecount = 0;
         int *clusterweights = (int *)malloc(sizeof(int)*clustercount);
         clustercount = 0;
         remainder = NULL;
         for (int s = 0; s < sclusterlist->size(); s++)
         {
            clusterweights[s] = dbs->clustersdb->GetClusterNodeWeight(sclusterlist->at(s));
            nodecount += clusterweights[s];
            clustercount++;
         }
         int tval = rand() % nodecount;
         int runtot = 0;
         int clusterchoice = 0;
         for (int j = 0; j < clustercount; j++)
         {
            runtot += clusterweights[j];
            if (runtot > tval)
            {
               clusterchoice = j;
               break;
            }
         }
         free(clusterweights);
         SSHChain *tchain = NULL;
         SSHClient *scli = NULL;
         std::vector<char *> *blacklist = new std::vector<char *>();
         std::vector<char *> *hostpath = dbs->clustersdb->GetPathFromSourceToCluster(final_user, sclusterlist->at(clusterchoice), blacklist);
         int fingerprintfailed = 0;
         while ((hostpath != NULL) && (tchain == NULL))
         {
            tchain = new SSHChain(dbs->clustersdb, mp);
            for (int i = 0; i < hostpath->size(); i++)
               tchain->AddToSSHChain(final_user, hostpath->at(i));
            tchain->SetAuthenticatorKeys(dbs->keydb, final_user, dbs->keydb->WIPpassword);
            int didfail;
            scli = tchain->ActivateSSHChain(&didfail);
            if (didfail)
            {
               char *failedhost = tchain->GetFailedLinkHostName();
               errprint("failed to access '%s', blacklisting...", failedhost);
               blacklist->push_back(failedhost);
               tchain->DustForFingerprints(dbs->pendingdb);
               fingerprintfailed = tchain->WasFingerprintProblem();
               delete tchain;
               tchain = NULL;
               StringUtils::destroyStringList(hostpath);
               hostpath = dbs->clustersdb->GetPathFromSourceToCluster(final_user, sclusterlist->at(clusterchoice), blacklist);
            }
         }
         StringUtils::destroyStringList(blacklist);
         if (tchain != NULL)
         {
            char *cluster_writedir = NULL;
            dbs->clustersdb->sqlGetSingleStringf(&cluster_writedir, "select writedir from 'clusters' where name='%s';", sclusterlist->at(clusterchoice));
            if ((cluster_writedir == NULL) || (strlen(cluster_writedir) == 0))
            {
               errprint("could not get cluster info for cluster named '%s'!", sclusterlist->at(clusterchoice));
               if (cluster_writedir) free(cluster_writedir);
               continue;
            }
            char *pubkey = rsakp->ExportPublicKey();
            char *prvkey = rsakp->ExportPrivateKey();
            char *pubkey_filename = StringUtils::varprintf("%s.pub", newuuid);
            char *prvkey_filename = StringUtils::varprintf("%s.prv", newuuid);
            if (scli->PutDataIntoSecureRemoteFile(cluster_writedir, pubkey_filename, pubkey, strlen(pubkey)) < 0)
            {
               errprint("could not copy public key to compute!");
               free(pubkey);
               free(prvkey);
               free(pubkey_filename);
               free(prvkey_filename);
               if (cluster_writedir) free(cluster_writedir);
               cluster_writedir = NULL;
               continue;
            }
            if (scli->PutDataIntoSecureRemoteFile(cluster_writedir, prvkey_filename, prvkey, strlen(prvkey)) < 0)
            {
               errprint("could not copy private key to compute!");
               free(pubkey);
               free(prvkey);
               free(pubkey_filename);
               free(prvkey_filename);
               if (cluster_writedir) free(cluster_writedir);
               cluster_writedir = NULL;
               continue;
            }
            // Adding ten minutes for construction and destruction and downloading/uploading from broker
            char *run_queueing_script = StringUtils::varprintf("export CONSPIRE_WRITEABLEDIR=%s; "
                                                               "export CONSPIRE_KEYID=%s; "
                                                               "export CONSPIRE_CLUSTERID=%s; "
                                                               "export CONSPIRE_TMAX=%d; "
                                                                  "(cd \"$CONSPIRE_WRITEABLEDIR\"; "
                                                                  "if [ `which wget` ]; then "
                                                                     "wget http://%s:%d/hosted_files/queue.sh -O queue_$CONSPIRE_KEYID.sh; "
                                                                  "else "
                                                                     "curl http://%s:%d/hosted_files/queue.sh > queue_$CONSPIRE_KEYID.sh; "
                                                                  "fi; "
                                                                  "chmod +x queue_$CONSPIRE_KEYID.sh; "
                                                                  "sh queue_$CONSPIRE_KEYID.sh; "
                                                                  "rm queue_$CONSPIRE_KEYID.sh)",
               cluster_writedir, newuuid, sclusterlist->at(clusterchoice), atoi(unresourced_work->at(i*3+2)) + 600,
               exthost, extport, exthost, extport);
//            %secho '(cd %s; if [ `which wget` ]; then wget http://%s:%d/hosted_files/bootstrap.sh -O bootstrap_%s.sh; else curl http://%s:%d/hosted_files/bootstrap.sh > bootstrap_%s.sh; fi; chmod +x bootstrap_%s.sh; sh bootstrap_%s.sh %s %d)' | %s",
//               first_command, clusterinfo->at(1), exthost, extport, newuuid, exthost, extport, newuuid, newuuid, newuuid, newuuid, 300, clusterinfo->at(0));
            printf("RUNNING QUEUE SCRIPT: %s\n", run_queueing_script);
            char *returnval = scli->RunCommand(run_queueing_script);
            if ((returnval == NULL) || (strlen(returnval) == 0))
            {
               errprint("received no response from cluster '%s' after attempting to sub '%s'", sclusterlist->at(clusterchoice), unresourced_work->at(i*3+0));
               free(pubkey);
               free(prvkey);
               free(pubkey_filename);
               free(prvkey_filename);
               free(run_queueing_script);
               if (cluster_writedir) free(cluster_writedir);
               cluster_writedir = NULL;
               continue;
            }
            delete tchain;
            errprint("CLUSTER JOB ID : '%s'\n", returnval);
            dbs->actorsdb->sqlInstantExecf("update 'workdata' set jobident='%s',jobclusterid='%s',computename='%s' where name='%s';",
               returnval, sclusterlist->at(clusterchoice), newuuid, unresourced_work->at(i*3+0));
            // Need to add lots of things written to databases saying that it has been tsarted
            // workdata clusterid and cluster... i forget
            free(pubkey);
            free(prvkey);
            free(pubkey_filename);
            free(prvkey_filename);
            free(run_queueing_script);
            if (cluster_writedir) free(cluster_writedir);
            cluster_writedir = NULL;
         } else
         {
            if ((hostpath == NULL) || (hostpath->size() == 0))
            {
               if (fingerprintfailed)
               {
                  errprint("problem with fingerprints caused a failure to schedule work '%s'", unresourced_work->at(i*3+0));
               } else
               {
                  errprint("problem with authentication or routing prevented work '%s' being scheduled", unresourced_work->at(i*3+0));
               }
            } else
            {
               char *prhlist = StringUtils::flattenStringList(hostpath, ",");
               if (fingerprintfailed)
               {
                  errprint("problem with fingerprints caused a failure to schedule work '%s' (list: '%s')", unresourced_work->at(i*3+0), prhlist);
               } else
               {
                  errprint("problem with authentication or routing prevented work '%s' being scheduled (list: '%s')", unresourced_work->at(i*3+0), prhlist);
               }
               free(prhlist);
               StringUtils::destroyStringList(hostpath);
               hostpath = NULL;
            }
         }
         if (hostpath) StringUtils::destroyStringList(hostpath);
         // Contents of sclusterlist
         free(clusterlist);
         // Actual string list
         delete sclusterlist;
         free(newuuid);
         delete rsakp;
         free(final_user);
      }
      StringUtils::destroyStringList(unresourced_work);
   }
}

void Broker4::UserRegistration(AllDBs *dbs, const char *clientname, BrokerMessage *sentmsg, BrokerMessage *replymsg)
{
   int response = DBArbitrator::CheckForUser(dbs, sentmsg->GetProcessee());
   //printf("Checking for user '%s'! - '%d'!\n", sentmsg->GetProcessee(), response);
   switch (response)
   {
      case CHECKSTATE_DOES_NOT_EXIST:
      {
         char timebuf[APR_RFC822_DATE_LEN + 1];
         apr_status_t status;
         apr_time_t timet = apr_time_now();
         status = apr_rfc822_date(timebuf, timet);
         CheckAPRError(status);
         // With the added new user request, also increment the broker key used - because it has a new user
         //dbs->keydb->IncrementKeyPairRefCount(sentmsg->GetRecipient());
         // Add SSH key into DB
         char *sshkeyowner = StringUtils::varprintf(".ssh_%s", sentmsg->GetProcessee());
         dbs->sshkeydb->AddPublicKey(sentmsg->GetProcessee(), "RSA", sentmsg->GetPayload(2), sshkeyowner);
         RSAKeyPair *rsakp = new RSAKeyPair();
         rsakp->ImportPrivateKey(sentmsg->GetPayload(3));
         dbs->sshkeydb->AddPrivateKeyWithPassword(sentmsg->GetProcessee(), sentmsg->GetPayload(1), (KeyPair *)rsakp, sshkeyowner);
         delete rsakp;
         // Create a table for the SSH logins 
         dbs->clustersdb->CreateUserTable(sentmsg->GetProcessee());
         // Add the public key to the broker
         dbs->keydb->AddPublicKey(sentmsg->GetProcessee(), "RSA", sentmsg->GetPayload(0), NULL);
         dbs->keydb->MarkKeyPairAsExpendable(sentmsg->GetProcessee());
         // Immediately deprecate the keys
         dbs->sshkeydb->TryToUpdateOwnersKeys(dbs->keydb, sshkeyowner, 0);
         char *latestsshkeyofowner = dbs->sshkeydb->GetLatestPublicKeyNameOfOwner(sshkeyowner);
         dbs->sshkeydb->MarkKeyPairAsExpendable(latestsshkeyofowner);
         free(latestsshkeyofowner);
         free(sshkeyowner);
         // Add actor as user
         dbs->actorsdb->AddUserToActors(sentmsg->GetProcessee(), timebuf, 0, 1024*1024*1024);
         char *latestownedkeyname = dbs->keydb->GetLatestPublicKeyNameOfOwner(sentmsg->GetProcessee());
         replymsg->SetRecipient(latestownedkeyname);
         free(latestownedkeyname);
         replymsg->AddTypeHeaderNOCRLF("REGISTER_USER_KEY", "AUTHORISED");
         break;
      }
      case CHECKSTATE_RESERVED:
      {
         char *latestownedkeyname = dbs->keydb->GetLatestPublicKeyNameOfOwner(sentmsg->GetProcessee());
         KeyPair *akp = dbs->keydb->GetPublicKeyByName(latestownedkeyname);
         free(latestownedkeyname);
         RSAKeyPair *bkp = new RSAKeyPair();
         bkp->ImportPublicKey(sentmsg->GetPayload(0));
         if (bkp->RSAKeyCompareThisPublicToPublic((RSAKeyPair *)akp))
         {
            // If the public keys match, readdress this to the user
            char *latestownedkeyname = dbs->keydb->GetLatestPublicKeyNameOfOwner(sentmsg->GetProcessee());
            replymsg->SetRecipient(latestownedkeyname);
            free(latestownedkeyname);
            replymsg->AddTypeHeaderNOCRLF("REGISTER_USER_KEY", "AUTHORISED");
         } else
         {
            errprint("named actor '%s' already exists with different key to that proffered by '%s'", sentmsg->GetProcessee(), clientname);
            dbs->keydb->DeletePublicKeyByName(sentmsg->GetSender());
            dbs->keydb->AddPublicKey(sentmsg->GetSender(), "RSA", sentmsg->GetPayload(0), NULL);
            replymsg->AddTypeHeaderNOCRLF("REGISTER_USER_KEY", "INVALID_NAME");
         }
         delete akp;
         delete bkp;
         break;
      }
      case CHECKSTATE_EXISTS_UNAUTHORISED:
      {
         dbs->keydb->DeletePublicKeyByName(sentmsg->GetSender());
         dbs->keydb->AddPublicKey(sentmsg->GetSender(), "RSA", sentmsg->GetPayload(0), NULL);
         replymsg->AddTypeHeaderNOCRLF("REGISTER_USER_KEY", "UNDER_REVIEW");
         break;
      }
   }
}

int Broker4::ProcessUploadedXMLDescriptor(AllDBs *dbs, const char *uuid, BrokerMessage *sentmsg, apr_off_t *sizeused)
{
   const apr_table_t *htable = sentmsg->GetHeaderTable();
   int retval = 0;
   const char *restype = apr_table_get(htable, "Session-Type");
   char *originated_from = NULL;
   if ((restype != NULL) && (strcmp(restype, "RESULTS") == 0))
   {
      const char *lident_str = apr_table_get(htable, "Local-Ident");
      if (lident_str) originated_from = StringUtils::varprintf("%s_%d", sentmsg->GetProcessee(), atoi(lident_str));
   }
   int fail_delete = 0;
   RSAKeyPair *jobkey = new RSAKeyPair();
   fail_delete |= (jobkey->GenerateKeyPair(KEY_BITS) != 0);
   dbs->keydb->AddPublicKey(uuid, jobkey, NULL);
   dbs->keydb->AddPrivateKeyWithPassword(uuid, dbs->keydb->WIPpassword, jobkey, NULL);
   delete jobkey;
   dbs->keydb->EnsureChainBaseAnchored(uuid);
   XMLDescriptor *xmldescr = new XMLDescriptor();
   xmldescr->readPlain(sentmsg->GetPayload(0), sentmsg->GetPayloadLength(0));
   char *jobxmlfilename = StringUtils::varprintf("_%s.xml", uuid);
   xmldescr->writeEncrypted(dbs->keydb, uuid, jobxmlfilename, sentmsg->GetEncMessageAssociatedMemoryPool());
   int process_errors = 0;
   if (xmldescr->allocateXMLDescribed(dbs, uuid, sentmsg->GetProcessee(),
          &process_errors, originated_from, sentmsg->GetEncMessageAssociatedMemoryPool(), sizeused))
   {
      errprint("Error during XML descriptor processing");
      fail_delete |= 1;
   }
   if (originated_from) free(originated_from);
   fail_delete |= (process_errors != 0);
   free(jobxmlfilename);
   delete xmldescr;
   return fail_delete;
}

void Broker4::CheckFileComplete(const char *sessionid, int blockno, BrokerMessage *replymsg)
{
   char *filename_c = StringUtils::varprintf("_%s.c%d", sessionid, blockno);
   apr_size_t fsize_c = Broker4::GetFileSize(filename_c, replymsg->GetEncMessageAssociatedMemoryPool());
   free(filename_c);
   char *filename_f = StringUtils::varprintf("_%s.f%d", sessionid, blockno);
   apr_size_t fsize_f = Broker4::GetFileSize(filename_f, replymsg->GetEncMessageAssociatedMemoryPool());
   free(filename_f);
   replymsg->AddTypeHeaderNOCRLF("FILECOMPLETE", (fsize_c > fsize_f) ? "N" : "Y");
}

void Broker4::WriteFileAndCheckHash(AllDBs *dbs, int blockno, BrokerMessage *sentmsg, BrokerMessage *replymsg)
{
   int result = 0;
   char *phash = Hash::getSHA1Hash(sentmsg->GetPayload(1), sentmsg->GetPayloadLength(1));
   if (strcmp(phash, sentmsg->GetPayload(0)) != 0) { errprint("data corruption - hashes do not match"); free(phash); return; }
   free(phash);
   char *filename_c = StringUtils::varprintf("_%s.c%d", sentmsg->GetProcessee(), blockno);
   apr_size_t fsize_c = Broker4::GetFileSize(filename_c, sentmsg->GetEncMessageAssociatedMemoryPool());
   free(filename_c);
   char *xmlfile = StringUtils::varprintf("_%s.xml", sentmsg->GetProcessee());
   XMLDescriptor *xmldescr = new XMLDescriptor();
   xmldescr->readEncrypted(dbs->keydb, sentmsg->GetProcessee(), xmlfile, sentmsg->GetEncMessageAssociatedMemoryPool());
   int cesize = xmldescr->getCompressedEncryptedSize(blockno);
   free(xmlfile);
   if (cesize == sentmsg->GetPayloadLength(1))
   {
      apr_status_t status;
      sentmsg->TakeAndWriteOutFileByName(1, "_%s.c%d", sentmsg->GetProcessee(), blockno);
      char *filename_c = StringUtils::varprintf("_%s.c%d", sentmsg->GetProcessee(), blockno);
      char *filename_f = StringUtils::varprintf("_%s.f%d", sentmsg->GetProcessee(), blockno);
      status = apr_file_rename(filename_c, filename_f, sentmsg->GetEncMessageAssociatedMemoryPool());
      free(filename_c); free(filename_f);
      CheckAPRError(status);
   }
   else
      errprint("payload size and size in xml descriptor do not match: payload %d, descriptor says %d", sentmsg->GetPayloadLength(1), cesize);
   delete xmldescr;
   Broker4::CheckFileComplete(sentmsg->GetProcessee(), blockno, replymsg);
}

#define MAX_BLOCK_SIZE_FILE (1048576*64)

void Broker4::WriteFakeFile(const char *filename, apr_size_t length, int *errmsg, int *mustcancel, apr_pool_t *mp)
{
   if (length > MAX_BLOCK_SIZE_FILE/*64Mb*/)
   {
      errprint("Block too big (size '%d' but maximum is set to '%d')", length, MAX_BLOCK_SIZE_FILE);
      *mustcancel = 1;
      if (errmsg) *errmsg = PROCESSXMLDESCRIPTOR_BLOCKTOOBIG;
      return;
   }
   apr_file_t *newFP;
   apr_status_t status;
   char *fakebuffer = (char *)malloc(sizeof(char)*length);
   for (int i = 0; i < length; i++) fakebuffer[i] = 0xFF;
   status = apr_file_open(&newFP, (const char *)filename, APR_FOPEN_CREATE|APR_FOPEN_WRITE|APR_FOPEN_BUFFERED|APR_FOPEN_BINARY, APR_OS_DEFAULT, mp);
   CheckAPRError(status);
   apr_size_t newsize = length;
   status = apr_file_write_full(newFP, fakebuffer, length, &newsize);
   CheckAPRError(status);
   status = apr_file_close(newFP);
   CheckAPRError(status);
   if (length != newsize)
   {
      errprint("Could not write data (bytes written %" APR_SIZE_T_FMT " out of %" APR_SIZE_T_FMT ")", newsize, length);
      *mustcancel = 1;
      if (errmsg) *errmsg = PROCESSXMLDESCRIPTOR_COULDNOTWRITE;
   }
   free(fakebuffer);
}

apr_size_t Broker4::GetFileSize(const char *filename, apr_pool_t *mp)
{
   apr_file_t *newFP;
   apr_status_t status;
   status = apr_file_open(&newFP, (const char *)filename, APR_FOPEN_READ|APR_FOPEN_BUFFERED|APR_FOPEN_BINARY, APR_OS_DEFAULT, mp);
   if (status != APR_SUCCESS)
   {
      return 0;
   }
//   CheckAPRError(status);
   apr_finfo_t finfo;
#ifdef USE_STAT_FILE_SIZE
   status = apr_file_info_get(&finfo, APR_FINFO_SIZE, newFP);
   CheckAPRError(status);
#else
   apr_off_t offset = 0;
   status = apr_file_seek(newFP, APR_END, &offset);
   CheckAPRError(status);
   finfo.size = offset;
   offset = 0;
   status = apr_file_seek(newFP, APR_SET, &offset);
   CheckAPRError(status);
#endif
   status = apr_file_close(newFP);
   CheckAPRError(status);
   return finfo.size;
}

std::vector<char> *Broker4::EncryptDataChunk(const char *cname, const char *sname, KeyPair *rsa, KeyPair *signing, const char *datachunk, int length, int dodecrypt, int docompress, apr_pool_t *mp, int *trusted)
{
   CryptEChain *crypty = new CryptEChain();
   crypty->buildBasicChain(cname, sname, rsa, signing, dodecrypt, docompress);
   APRBucketBrigade *bucketbrgde = new APRBucketBrigade(mp);
   int smallblock = (ENV_READ_BUFFER > (length/2)) ? (length/2) : ENV_READ_BUFFER;
   for (int k = 0; k < length; k += smallblock)
   {
      int actualamt = ((length - k) < smallblock) ? length - k : smallblock;
      int bytes = bucketbrgde->writeToStream(&(datachunk[k]), actualamt);
      if (bytes == 0)
      {
         errprint("push to bucket brigade failed - incomplete data will be sent");
         break;
      }
      k -= (smallblock - bytes);
      // Check bytes, seriously!
   }
   bucketbrgde->writeEOS();
   APRBucketBrigade *outputbrgde = new APRBucketBrigade(mp);
   std::vector<char> *largefile = new std::vector<char>();
   largefile->reserve(length);
   int proc = 0;
   while (!proc)
   {
      proc = crypty->processIteration(bucketbrgde, outputbrgde);
      apr_size_t reallen = outputbrgde->realLength();
      if (reallen > 0)
      {
         char *alength = (char *)malloc(sizeof(char)*(reallen+1));
         apr_size_t gotreal = outputbrgde->popFromStream(alength, reallen);
         for (int ii = 0; ii < (int)gotreal; ii++)
            largefile->push_back(alength[ii]);
         free(alength);
      }
   }
   delete bucketbrgde;
   delete outputbrgde;
   if (trusted) *trusted = crypty->isTrusted();
   delete crypty;
   return largefile;
}

std::vector<char> *Broker4::EncryptFileBlock(const char *cname, const char *sname, KeyPair *rsa, KeyPair *signing, const char *filename, apr_size_t rangebegin, apr_size_t rangeend, int dodecrypt, int docompress, apr_pool_t *mp, int *trusted)
{
   CryptEChain *crypty = new CryptEChain();
   crypty->buildBasicChain(cname, sname, rsa, signing, dodecrypt, docompress);
   APRBucketBrigade *bucketbrgde = new APRBucketBrigade(mp);
   bucketbrgde->writeFileToStream(filename, rangebegin, rangeend, 1);
   bucketbrgde->writeEOS();
   APRBucketBrigade *outputbrgde = new APRBucketBrigade(mp);
   std::vector<char> *largefile = new std::vector<char>();
   largefile->reserve(rangeend - rangebegin);
   int proc = 0;
   while (!proc)
   {
      proc = crypty->processIteration(bucketbrgde, outputbrgde);
      apr_size_t reallen = outputbrgde->realLength();
      if (reallen > 0)
      {
         char *alength = (char *)malloc(sizeof(char)*(reallen+1));
         apr_size_t gotreal = outputbrgde->popFromStream(alength, reallen);
         for (int ii = 0; ii < (int)gotreal; ii++)
            largefile->push_back(alength[ii]);
         free(alength);
      }
   }
   delete bucketbrgde;
   delete outputbrgde;
   if (trusted) *trusted = crypty->isTrusted();
   delete crypty;
   return largefile;
}
